﻿#include "coro_runner_impl.hh"
#include "../../src/repo/src/include/root/coroutine/coroutine.h"
#include "../box/service_box.hh"
#include "../detail/lang_impl.hh"
#include "../detail/service_context_impl.hh"
#include <list>
#include <stdexcept>
#include <thread>

kratos::service::CoroRunnerImpl::CoroRunnerImpl(ServiceContextImpl *ctx) {
  ctx_ = ctx;
}

kratos::service::CoroRunnerImpl::~CoroRunnerImpl() {
  // 销毁时保证销毁所有未退出的协程，防止组件被卸载时未退出协程造成的内存泄露
  stop_all_co();
}

auto kratos::service::CoroRunnerImpl::start_co(kratos::service::CoFunction func)
    -> COID {
  if (!func) {
    return 0;
  }
  auto id = coro_start([&](void *) {
    // 保证持有的CoroRunner的生命周期大于协程的生命周期
    auto id = coro_id();
    coro_set_.emplace(id);
    run(func);
    coro_set_.erase(id);
  });
  return id;
}

auto kratos::service::CoroRunnerImpl::sleep_co(std::time_t ms) -> void {
  if (coro_is_main()) {
    return;
  }
  coro_sleep(ms);
}

auto kratos::service::CoroRunnerImpl::yield_co() -> void {
  if (coro_is_main()) {
    return;
  }
  return coro_yield();
}

auto kratos::service::CoroRunnerImpl::resume_co(COID coid) -> void {
  coro_resume(coid);
}

auto kratos::service::CoroRunnerImpl::try_co() -> std::any {
  if (coro_is_main()) {
    return std::any();
  }
  return coro_pop();
}

auto kratos::service::CoroRunnerImpl::wait_co() -> std::any {
  if (coro_is_main()) {
    return std::any();
  }
  return coro_pop_wait();
}

auto kratos::service::CoroRunnerImpl::wait_co(std::time_t timeout) -> std::any {
  if (coro_is_main()) {
    return std::any();
  }
  return coro_pop_wait(timeout);
}

auto kratos::service::CoroRunnerImpl::await_cancel_co(std::time_t timeout,
                                                      CoFunction func) -> bool {
  if (coro_is_main()) {
    return false;
  }
  if (!func) {
    return false;
  }
  if (!timeout) {
    timeout = 1;
  }
  auto parent_id = get_id();
  auto sub_id = coro_spawn([&,parent_id](void *) {
    auto id = coro_id();
    run(func);
    coro_set_.erase(id);
    resume_co(parent_id);
  });
  coro_set_.emplace(sub_id);
  coro_wakeup(sub_id);
  wait_co(timeout);
  // 将在新协程及其之后产生的所有协程都取消
  std::list<COID> id_list;
  for (const auto &id : coro_set_) {
    if (id > parent_id) {
      id_list.push_back(id);
    }
  }
  if (!id_list.empty()) {
    // 最终剩余的协程数量
    auto reserve_count = coro_set_.size() - id_list.size();
    // 按照协程ID降序排列，越大的ID越晚建立
    id_list.sort(std::greater<COID>());
    // 所有没有正常退出的协程都取消
    for (auto id : id_list) {
      cancel_co(id);
    }
  }
  return true;
}

auto kratos::service::CoroRunnerImpl::push_co(COID coid, std::any &data)
    -> bool {
  coro_push(coid, data);
  return true;
}

auto kratos::service::CoroRunnerImpl::cancel_co(COID coid) -> void {
  coro_cancel(coid);
  coro_wakeup(coid);
}

auto kratos::service::CoroRunnerImpl::stop_all_co() -> void {
  auto temp_set = coro_set_;
  for (const auto &id : temp_set) {
    cancel_co(id);
  }
  coro_set_.clear();
}

auto kratos::service::CoroRunnerImpl::get_id() -> COID { return coro_id(); }

void kratos::service::CoroRunnerImpl::run(CoFunction cof) {
  try {
    cof();
  } catch (CoroCancelException &) {
    // 被强制取消运行
  } catch (std::exception &ex) {
    if (ctx_) {
      ctx_->get_box()->write_log(lang::LangID::LANG_CORO_RUNNER_EXCEPTION,
                                 klogger::Logger::EXCEPTION, ex.what());
    }
  }
}
